/**
 * Software License, Version 1.0
 * 
 * Copyright 2003 The Trustees of Indiana University.  All rights reserved.
 * 
 *
 *Redistribution and use in source and binary forms, with or without 
 *modification, are permitted provided that the following conditions are met:
 *
 *1) All redistributions of source code must retain the above copyright notice,
 * the list of authors in the original source code, this list of conditions and
 * the disclaimer listed in this license;
 *2) All redistributions in binary form must reproduce the above copyright 
 * notice, this list of conditions and the disclaimer listed in this license in
 * the documentation and/or other materials provided with the distribution;
 *3) Any documentation included with all redistributions must include the 
 * following acknowledgement:
 *
 *"This product includes software developed by the Community Grids Lab. For 
 * further information contact the Community Grids Lab at 
 * http://communitygrids.iu.edu/."
 *
 * Alternatively, this acknowledgement may appear in the software itself, and 
 * wherever such third-party acknowledgments normally appear.
 * 
 *4) The name Indiana University or Community Grids Lab or NaradaBrokering, 
 * shall not be used to endorse or promote products derived from this software 
 * without prior written permission from Indiana University.  For written 
 * permission, please contact the Advanced Research and Technology Institute 
 * ("ARTI") at 351 West 10th Street, Indianapolis, Indiana 46202.
 *5) Products derived from this software may not be called NaradaBrokering, 
 * nor may Indiana University or Community Grids Lab or NaradaBrokering appear
 * in their name, without prior written permission of ARTI.
 * 
 *
 * Indiana University provides no reassurances that the source code provided 
 * does not infringe the patent or any other intellectual property rights of 
 * any other entity.  Indiana University disclaims any liability to any 
 * recipient for claims brought by any other entity based on infringement of 
 * intellectual property rights or otherwise.  
 *
 *LICENSEE UNDERSTANDS THAT SOFTWARE IS PROVIDED "AS IS" FOR WHICH NO 
 *WARRANTIES AS TO CAPABILITIES OR ACCURACY ARE MADE. INDIANA UNIVERSITY GIVES
 *NO WARRANTIES AND MAKES NO REPRESENTATION THAT SOFTWARE IS FREE OF 
 *INFRINGEMENT OF THIRD PARTY PATENT, COPYRIGHT, OR OTHER PROPRIETARY RIGHTS. 
 *INDIANA UNIVERSITY MAKES NO WARRANTIES THAT SOFTWARE IS FREE FROM "BUGS", 
 *"VIRUSES", "TROJAN HORSES", "TRAP DOORS", "WORMS", OR OTHER HARMFUL CODE.  
 *LICENSEE ASSUMES THE ENTIRE RISK AS TO THE PERFORMANCE OF SOFTWARE AND/OR 
 *ASSOCIATED MATERIALS, AND TO THE PERFORMANCE AND VALIDITY OF INFORMATION 
 *GENERATED USING SOFTWARE.
 */
package cgl.narada.service.reliable.impl;

import java.util.LinkedList;

import cgl.narada.event.NBEvent;
import cgl.narada.event.NBEventTypes;
import cgl.narada.service.ServiceException;
import cgl.narada.service.reliable.events.RdAckInvoiceEntityEvent;
import cgl.narada.service.reliable.events.RdAckResponseServiceInvoiceEvent;
import cgl.narada.service.reliable.events.RdArchivalServiceNotification;
import cgl.narada.service.reliable.events.RdCompanionEntityEvent;
import cgl.narada.service.reliable.events.RdNakInvoiceEntityEvent;
import cgl.narada.service.reliable.events.RdNegotiationsServiceEvent;
import cgl.narada.service.reliable.events.RdProfileUpdateRequest;
import cgl.narada.service.reliable.events.RdProfileUpdateResponse;
import cgl.narada.service.reliable.events.RdRecoveryEntityRequest;
import cgl.narada.service.reliable.events.RdRecoveryResponse;
import cgl.narada.service.reliable.events.RdRepublishedEntityEvent;
import cgl.narada.service.reliable.events.RdRetransmissionServiceEvent;
import cgl.narada.service.reliable.events.ReliableDeliveryExchange;

/** This class is responsible for reconstructing the byte streams and 
 rerouting the events to the corresponding handlers.
 
 @author Shrideep Pallickara
 $Date$
 $Revision$

 */

public class RdCommunicationsMultiplexerImpl extends Thread {
	private ReliableDeliveryServiceImpl	reliableDeliveryService	= null;
	private RdPublishingEntityImpl			rdPublisher							= null;
	private SubscriberPercolator				subscriberPercolator		= null;
	private boolean											keepLooping							= true;
	private boolean											debug										= false;

	//  private Vector receivedEvents;
	private LinkedList									receivedEvents;
	private Object											syncObject;
	private boolean											waiting									= true;

	private String											moduleName							= "RdCommunicationsMultiplexerImpl: ";

	public RdCommunicationsMultiplexerImpl() {
		syncObject = new Object();
		//    receivedEvents = new Vector();
		receivedEvents = new LinkedList();
		start();
		setName("RdCommunicationsMultiplexer");
	}

	public void registerReliableDeliveryService(
			ReliableDeliveryServiceImpl rdsImpl) throws ServiceException {
		if (reliableDeliveryService != null) {
			throw new ServiceException(moduleName + "Reliable Delivery Service"
					+ " previously registered");
		}

		reliableDeliveryService = rdsImpl;

	}

	public void registerSubscriberPercolator(
			SubscriberPercolator subscriberPercolator) {
		this.subscriberPercolator = subscriberPercolator;
	}

	public void registerPublishingEntity(RdPublishingEntityImpl rdPublisherImpl)
			throws ServiceException {
		if (rdPublisher != null) {
			throw new ServiceException(moduleName + "Reliable Delivery publisher"
					+ " previously registered");
		}
		this.rdPublisher = rdPublisherImpl;
	}

	public void registerRetransmissionService() {

	}

	public void registerRecoveryService() {

	}

	public void onEvent(NBEvent nbEvent) {
		int eventType = nbEvent.getEventType();

		if (eventType == NBEventTypes.NORMAL_EVENT) {
			reliableDeliveryService.processPublishedEvent(nbEvent);
			return;
		}

		if (eventType == NBEventTypes.RELIABLE_DELIVERY_EXCHANGE) {
			manageReliableDeliveryExchange(nbEvent);
			return;
		}

		System.out.println(moduleName + "Unknown event type [" + eventType
				+ "] received ");
	}

	public void manageReliableDeliveryExchange(NBEvent nbEvent) {
		//    receivedEvents.addElement(nbEvent);
		receivedEvents.add(nbEvent);
		Thread.yield();
		if (waiting && (receivedEvents.size() > 0)) {
			synchronized (syncObject) {
				syncObject.notify();
			}
			;
		}
	}

	public void run() {
		try {
			while (keepLooping) {
				if (receivedEvents.size() == 0) {
					synchronized (syncObject) {
						waiting = true;
						if (debug) {
							System.out.println(moduleName + "Waiting for notification");
						}

						syncObject.wait();

						if (debug)
							System.out.println(moduleName + "Waking up!");
						waiting = false;
					}
					;/*end synchronized(syncObject) */
				}/* end if receivedEvents.size() == 0 */

				if (receivedEvents.size() == 0) {
					continue;
				}

				//	NBEvent eventToProcess = (NBEvent)receivedEvents.elementAt(0);
				//	multiplexReliableDeliveryExchange(eventToProcess);
				//	receivedEvents.removeElementAt(0);

				NBEvent eventToProcess = (NBEvent) receivedEvents.removeFirst();
				multiplexReliableDeliveryExchange(eventToProcess);

			}/*end (while) */
		} catch (Exception e) {
			e.printStackTrace();
		}

	}

	/** Terminating the service */
	public void terminateService() throws ServiceException {
		keepLooping = false;
		synchronized (syncObject) {
			syncObject.notify();
		}
		;
	}

	public void multiplexReliableDeliveryExchange(NBEvent nbEvent) {
		byte[] contentPayload = nbEvent.getContentPayload();

		byte exchangeType = contentPayload[0];

		if (exchangeType == ReliableDeliveryExchange.COMPANION_ENTITY_EVENT) {
			RdCompanionEntityEvent companionEvent = new RdCompanionEntityEvent(
					contentPayload);

			//			System.out.println(moduleName + "Received companion event ... "
			//					+ companionEvent.getCatenationNumber()+"\n\t Event ID:"+companionEvent.getEventId());

			if (reliableDeliveryService != null) {
				reliableDeliveryService.processCompanionEvent(companionEvent);
			} else {
				System.out.println(moduleName + "Received companion event ... "
						+ "no registered reliable delivery service");
			}
			return;
		}

		if (exchangeType == ReliableDeliveryExchange.REPUBLISHED_ENTITY_EVENT) {
			RdRepublishedEntityEvent republishedEvent = new RdRepublishedEntityEvent(
					contentPayload);
			if (reliableDeliveryService != null) {
				System.out.println(moduleName + "Received republished event");
				reliableDeliveryService.processRepublishedEvent(republishedEvent);
			} else {
				System.out.println(moduleName + "Received republished event BUT "
						+ "NO registered reliable delivery service");
			}
			return;
		}

		if (exchangeType == ReliableDeliveryExchange.RDS_ENTITY_NEGOTIATIONS) {
			RdNegotiationsServiceEvent negotiationsEvent = new RdNegotiationsServiceEvent(
					contentPayload);
			if (rdPublisher != null) {
				rdPublisher.processNegotiationServiceEvent(negotiationsEvent);
			} else {
				System.out.println(moduleName + "Received negotiation event, but "
						+ "there are no registered RDPublishing Entities!");
			}
			return;
		}

		if (exchangeType == ReliableDeliveryExchange.ENTITY_RDS_ACK_INVOICE_EVENT) {
			RdAckInvoiceEntityEvent ackInvoiceEvent = new RdAckInvoiceEntityEvent(
					contentPayload);
			if (reliableDeliveryService != null) {
				if (debug) {
					System.out.println(moduleName + "Received ack invoice event");
				}
				try {
					reliableDeliveryService.processAckInvoiceEvent(ackInvoiceEvent);
				} catch (ServiceException serEx) {
					System.out.println(moduleName + serEx);
				}

			} else {
				System.out.println(moduleName + "Received ack invoice event BUT "
						+ "NO registered reliable delivery service");
			}
			return;
		}

		if (exchangeType == ReliableDeliveryExchange.ENTITY_RDS_NAK_INVOICE_EVENT) {
			RdNakInvoiceEntityEvent nakInvoiceEvent = new RdNakInvoiceEntityEvent(
					contentPayload);
			if (reliableDeliveryService != null) {
				System.out.println(moduleName + "Received nak invoice event");
				try {
					reliableDeliveryService.processNakInvoiceEvent(nakInvoiceEvent);
				} catch (ServiceException serEx) {
					System.out.println(moduleName + serEx);
				}
			} else {
				System.out.println(moduleName + "Received nak invoice event BUT "
						+ "NO registered reliable delivery service");
			}
			return;
		}

		if (exchangeType == ReliableDeliveryExchange.ENTITY_RECOVERY_REQUEST) {
			RdRecoveryEntityRequest recoveryRequest = new RdRecoveryEntityRequest(
					contentPayload);
			if (reliableDeliveryService != null) {
				System.out.println(moduleName + "Received entity recovery request");
				try {
					reliableDeliveryService.processEntityRecoveryRequest(recoveryRequest);
				} catch (ServiceException serEx) {
					System.out.println(moduleName + serEx);
				}
			} else {
				System.out.println(moduleName + "Received entity recovery request BUT "
						+ "NO registered reliable delivery service");
			}
			return;
		}

		if (exchangeType == ReliableDeliveryExchange.ENTITY_PROFILE_UPDATE_REQUEST) {
			RdProfileUpdateRequest profileRequest = new RdProfileUpdateRequest(
					contentPayload);
			if (reliableDeliveryService != null) {
				System.out.println(moduleName + "Received profile update request");
				try {
					reliableDeliveryService.processProfileUpdateRequest(profileRequest);
				} catch (ServiceException serEx) {
					System.out.println(moduleName + serEx);
				}
			} else {
				System.out.println(moduleName + "Received profile update request BUT "
						+ "NO registered reliable delivery service");
			}
			return;
		}

		if (exchangeType == ReliableDeliveryExchange.ARCHIVAL_NOTIFICATION) {
			RdArchivalServiceNotification archivalNotification = new RdArchivalServiceNotification(
					contentPayload);
			if (subscriberPercolator != null) {
				try {
					subscriberPercolator
							.processArchivalNotification(archivalNotification);
				} catch (ServiceException serEx) {
					System.out.println(moduleName + serEx);
				}

			} else {
				System.out.println(moduleName + "Received archival notification BUT"
						+ " NO registered entity actuator available");
			}

			return;
		}

		if (exchangeType == ReliableDeliveryExchange.RDS_ACK_RESPONSE_INVOICE_EVENT) {
			RdAckResponseServiceInvoiceEvent ackResponseInvoiceEvent = new RdAckResponseServiceInvoiceEvent(
					contentPayload);
			if (subscriberPercolator != null) {
				try {
					subscriberPercolator.processAckResponseEvent(ackResponseInvoiceEvent);
				} catch (ServiceException serEx) {
					System.out.println(moduleName + serEx);
				}

			} else {
				System.out.println(moduleName + "Received archival notification BUT"
						+ " NO registered entity actuator available");
			}
			return;
		}

		if (exchangeType == ReliableDeliveryExchange.RETRANSMITTED_RDS_EVENT) {
			RdRetransmissionServiceEvent retransmittedEvent = new RdRetransmissionServiceEvent(
					contentPayload);
			if (subscriberPercolator != null) {
				try {
					subscriberPercolator.processRetransmittedEvent(retransmittedEvent);
				} catch (ServiceException serEx) {
					System.out.println(moduleName + serEx);
				}

			} else {
				System.out.println(moduleName + "Received archival notification BUT"
						+ " NO registered entity actuator available");
			}
			return;
		}

		if (exchangeType == ReliableDeliveryExchange.ENTITY_RECOVERY_RESPONSE) {
			RdRecoveryResponse recoveryResponse = new RdRecoveryResponse(
					contentPayload);

			boolean isForPublisherPercolator = recoveryResponse
					.hasPublisherRecoveryInfo();

			if (isForPublisherPercolator) {
				if (rdPublisher != null) {
					rdPublisher.processRecoveryResponse(recoveryResponse);
				} else {
					System.out.println(moduleName + "Received recovery response, but "
							+ "there are no registered RDPublishing Entities!");
				}

			} else {
				if (subscriberPercolator != null) {
					try {
						subscriberPercolator.processRecoveryResponse(recoveryResponse);
					} catch (ServiceException serEx) {
						System.out.println(moduleName + serEx);
					}

				} else {
					System.out.println(moduleName + "Received recovery response BUT NO "
							+ "registered subscriber percolator available");
				}
			}
			/** end (isForPublisherPercolator) */
			return;
		}

		if (exchangeType == ReliableDeliveryExchange.RDS_PROFILE_UPDATE_RESPONSE) {

			RdProfileUpdateResponse profileUpdateResp = new RdProfileUpdateResponse(
					contentPayload);
			if (subscriberPercolator != null) {
				try {
					subscriberPercolator.processProfileUpdateResponse(profileUpdateResp);
				} catch (ServiceException serEx) {
					System.out.println(moduleName + serEx);
				}

			} else {
				System.out.println(moduleName + "Received profile update response BUT"
						+ " NO registered entity actuator available");
			}
			return;
		}

		System.out.println(moduleName + "Unknown Reliable Delivery Exchange ["
				+ exchangeType + "] received");
	}

}
